package com.shujia.sql

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.functions.ScalarFunction

import java.time.LocalDateTime

object Demo05ProcessTime {
  def main(args: Array[String]): Unit = {
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() // 使用Blink的SQL解析器
      .inStreamingMode() // 开启流处理模式
      .build()
    // 构建Table Env
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    bsEnv.setParallelism(1)

    bsTableEnv.createTemporaryFunction("add8hours", new Add8Hours)


    // 构建Source表 动态表
    bsTableEnv
      .executeSql(
        """
          |CREATE TABLE click_table (
          | `user` STRING,
          | url STRING,
          | procT AS PROCTIME()  -- Flink SQL在使用处理时间时没有考虑时区问题
          |) WITH (
          | 'connector' = 'kafka',
          | 'topic' = 'click2',
          | 'properties.bootstrap.servers' = 'master:9092,node1:9092,node2:9092',
          | 'properties.group.id' = 'test1',
          | 'format' = 'csv',
          | 'csv.field-delimiter' = ',',
          | 'csv.ignore-parse-errors' = 'true',
          | 'scan.startup.mode' = 'earliest-offset'
          |)
          |""".stripMargin)


    bsTableEnv
      .executeSql(
        """
          |create table print_table(
          | `user` String,
          | cnt BIGINT,
          | startT TIMESTAMP(3),
          | endT TIMESTAMP(3)
          |)
          |with (
          | 'connector' = 'print'
          |)
          |""".stripMargin)

    // 每隔5s统计最近10s内的每个用户的点击次数
    // 滑动窗口：
    //        滑动时间：5s
    //        窗口大小：10s

    bsTableEnv
      .executeSql(
        """
          |insert into print_table
          |select user
          |       ,count(*) as cnt
          |       ,add8hours(HOP_START(procT, INTERVAL '5' SECONDS ,  INTERVAL '10' SECONDS)) as startT
          |       ,add8hours(HOP_END(procT, INTERVAL '5' SECONDS ,  INTERVAL '10' SECONDS)) as endT
          |from click_table
          |group by user,
          |HOP(procT, INTERVAL '5' SECONDS ,  INTERVAL '10' SECONDS)
          |""".stripMargin)


  }

}

class Add8Hours extends ScalarFunction {
  def eval(time: LocalDateTime): LocalDateTime = {
    // 将时间加8个小时并返回
    time.plusHours(8)
  }
}